package com.harboursoftware.xstorage.unit.fs

;

import junit.framework.Assert;

import org.apache.commons.io.IOUtils;
import org.apache.commons.io.input.AutoCloseInputStream 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;


import com.harboursoftware.xstorage.ac.AccessControlPolicy 
import com.harboursoftware.xstorage.db.BucketDao 
import com.harboursoftware.xstorage.fs.BucketAlreadyOwnedByOtherUserException 
import com.harboursoftware.xstorage.fs.BucketNotEmptyException 
import com.harboursoftware.xstorage.fs.BucketNotExistException 
import com.harboursoftware.xstorage.fs.HDFSFileSystem 
import com.harboursoftware.xstorage.fs.ObjectNotExistException 
import com.harboursoftware.xstorage.model.XBucket 
import com.harboursoftware.xstorage.model.XObject 
import com.harboursoftware.xstorage.model.XUser 
import com.harboursoftware.xstorage.unit.mock.MockBucketDao 



/**
 * TODO ：重构该类  
 */
public class HDFSFileSystemTest {
    HDFSFileSystem fileSystem;
    
    FileSystem localHadoopFileSystem = FileSystem.getLocal(new Configuration(true))
    BucketDao mockBucketDao = new MockBucketDao()
	def user = new XUser(id : 'testuser')
	def bucketName = 'testbucket'
    
    @Before
    void setUp() throws Exception {
        fileSystem = new HDFSFileSystem(localHadoopFileSystem, mockBucketDao);
    }
    
    @After
    void tearDown() throws Exception {
        //clear 
        localHadoopFileSystem.delete(new Path('/' + bucketName), true);     
        mockBucketDao.bucketMap.clear()
        fileSystem = null
    }   
    
    @Test
    void test_createBucket() throws Exception {
        def owner = new XUser(id : 'testuser')
		def acp = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner, owner);
        def bucket = new XBucket(name : bucketName, owner : owner, acp : acp)
        fileSystem.createBucket(bucket)
        
        def pathString = Path.SEPARATOR + bucketName
        def path = new Path(pathString)
        assert localHadoopFileSystem.exists(path)
        
        def persistentedBucket = mockBucketDao.findBucketByName(bucketName)
        assert bucket == persistentedBucket
    }
	
	
	@Test(expected = BucketAlreadyOwnedByOtherUserException.class)
	void test_createBucket_when_bucket_already_eixst() throws Exception {
		XUser owner1 = new XUser(id : 'testuser1')
		def acp1 = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner1, owner1)
		XBucket bucket1 = new XBucket(name : bucketName, owner : owner1, acp : acp1)
		XUser owner2 = new XUser(id : 'testuser2')
		def acp2 = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner2, owner2)
		XBucket bucket2 = new XBucket(name : bucketName, owner : owner2)
		
		fileSystem.createBucket(bucket1)
		fileSystem.createBucket(bucket2)
	}
	
    /**
     * will update acp of bucket
     */
    @Test
	void test_createBucket_when_bucket_already_own_by_same_user() throws Exception {
		def acp1 = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, user, user)
		def bucket1 = new XBucket(name : bucketName, owner : user, acp : acp1)
		fileSystem.createBucket(bucket1);
		
		def acp2 = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PUBLIC_READE, user, user)
		def bucket2 = new XBucket(name : bucketName, owner : user, acp : acp2)
		
		fileSystem.createBucket bucket2
        fileSystem.loadBucketACP bucket2

        Assert.assertEquals acp2, bucket2.acp
	}

	@Test
	void test_deleteBucket() throws Exception {
		def owner = new XUser(id : 'testuser')
		def acp = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner, owner);
		def bucket = new XBucket(name : bucketName, owner : owner, acp : acp)
		fileSystem.createBucket(bucket)
		fileSystem.deleteBucket(bucket)
		
        def pathString = Path.SEPARATOR + bucketName
        def path = new Path(pathString)
        assert !localHadoopFileSystem.exists(path)
        assert mockBucketDao.findBucketByName(bucketName) == null
	}
    
    @Test(expected = BucketNotEmptyException)
    void test_deleteBucket_when_bucket_not_empty() {
        //create bucket
        XBucket bucket = new XBucket(name : bucketName)
        fileSystem.createBucket bucket
        
        //create object
        def key = '/a'
        def input = IOUtils.toInputStream('message digest')
        def object = new XObject(key : key, bucket : bucket, inputStream : input)
        fileSystem.createObject object
        
        fileSystem.deleteBucket bucket
    }

	@Test(expected = BucketNotExistException.class)
	void test_deleteBucket_when_bucket_not_exist() throws Exception {
		XBucket bucket = new XBucket(name : 'sdafadsfsdaklfjkldsa')
		fileSystem.deleteBucket(bucket)
	}

	@Test
	void test_loadBucketACP() {
		def owner = new XUser(id : 'testuser')
		def acp = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner, owner);
		def bucket = new XBucket(name : bucketName, owner : owner, acp : acp)
		fileSystem.createBucket(bucket)

		def newBucket = new XBucket(name : bucketName)
		fileSystem.loadBucketACP newBucket
		assert bucket.acp == newBucket.acp
	}
	
	@Test(expected = BucketNotExistException.class)
    void test_loadBucketACP_after_delete_bucket() {
        def owner = new XUser(id : 'testuser')
        def acp = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner, owner);
        def bucket = new XBucket(name : bucketName, owner : owner, acp : acp)
        fileSystem.createBucket(bucket)
        fileSystem.deleteBucket(bucket)
        
        assert mockBucketDao.findBucketByName(bucketName) == null
        
        def bucket2 = new XBucket(name : bucketName)
        fileSystem.loadBucketACP bucket2;
    }

	
	@Test
	void test_createObject() {
        //create bucket
		def owner = new XUser(id : 'testuser')
		def acp = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner, owner);
		def bucket = new XBucket(name : bucketName, owner : owner, acp : acp)
		fileSystem.createBucket bucket
		
        //create object
		def key = '/a'
		def input = IOUtils.toInputStream('message digest')
		def object = new XObject(key : key, bucket : bucket, inputStream : input)
		def md5Digest = fileSystem.createObject(object)
		assert md5Digest == 'f96b697d7cb7938d525a2f31aaf161d0'
		
		def path = new Path(Path.SEPARATOR + bucketName + key)
		def input1 = localHadoopFileSystem.open(path)
		def objectInputStream = new ObjectInputStream(input1)
		objectInputStream.readObject()
		objectInputStream.readObject()
		assert IOUtils.toString(new AutoCloseInputStream(objectInputStream)) == 'message digest'
	}
	
	@Test
	void test_loadObject() {
		//create bucket
		def owner = new XUser(id : 'testuser')
		def acp = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner, owner);
		def bucket = new XBucket(name : bucketName, owner : owner, acp : acp)
		fileSystem.createBucket bucket
		
		//create object
		def input = IOUtils.toInputStream('message digest')
		def object = new XObject(key : '/a', bucket : bucket, inputStream : input)
		fileSystem.createObject(object)

		//load object
		def newObject = new XObject(bucket : bucket, key : '/a')
		fileSystem.loadObject(newObject)

		//asserts
		assert newObject.acp == null
		assert newObject.headers == null
		assert IOUtils.toString(new AutoCloseInputStream(newObject.inputStream)) == 'message digest'
	}
	
	
	@Test
	void test_loadObject_2() {
		//create bucket
		def owner = new XUser(id : 'testuser')
		def acp = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner, owner);
		def bucket = new XBucket(name : bucketName, owner : owner, acp : acp)
		fileSystem.createBucket bucket
		
		//create object
		def input = IOUtils.toInputStream('message digest')
		def headers = ['Content_Type' : 'text/xml']
		def object = new XObject(key : '/a', bucket : bucket, inputStream : input, acp : acp, headers : headers)
		fileSystem.createObject(object)
		
		//get object
		def newObject = new XObject(bucket : bucket, key : '/a')
		fileSystem.loadObject(newObject)
		
		//asserts
		assert newObject.acp == acp
		assert newObject.headers == headers
		assert IOUtils.toString(new AutoCloseInputStream(newObject.inputStream)) == 'message digest'
	}
	
	@Test
	void test_loadObjectEtag() {
		//create bucket
		def owner = new XUser(id : 'testuser')
		def acp = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner, owner);
		def bucket = new XBucket(name : bucketName, owner : owner, acp : acp)
		fileSystem.createBucket bucket
		
		//create object
		def input = IOUtils.toInputStream('message digest')
		def object = new XObject(key : '/a', bucket : bucket, inputStream : input)
		fileSystem.createObject(object)

		//load object
		def newObject = new XObject(bucket : bucket, key : '/a')
		fileSystem.loadObjectEtag(newObject)

		//asserts
		assert newObject.etag == 'f96b697d7cb7938d525a2f31aaf161d0'
	}
	
	@Test(expected = BucketNotExistException.class)
	void test_getObject_when_bucket_not_exists() {
		def bucket = new XBucket(name : bucketName)
		def object = new XObject(bucket : bucket, key : '/a')
		fileSystem.loadObject(object)
	}
	
	@Test(expected = ObjectNotExistException.class)
	void test_getObject_when_object_not_exists() {
		//create bucket
		def owner = new XUser(id : 'testuser')
		def acp = AccessControlPolicy.createCannedAccessPolicy(AccessControlPolicy.CANNED_ACP_PRIVATE, owner, owner);
		def bucket = new XBucket(name : bucketName, owner : owner, acp : acp)
		fileSystem.createBucket bucket
		
		//get object
		def object = new XObject(bucket : bucket, key : '/a')
		fileSystem.loadObject(object)
	}
}
